package com.ubxtech.rocketmq.core;

import cn.hutool.core.util.StrUtil;
import com.ubxtech.core.context.ContextConstants;
import com.ubxtech.core.context.ContextUtil;
import com.ubxtech.core.dto.MsgDTO;
import com.ubxtech.rocketmq.properties.RocketMqConfigProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils;

/**
 * rocketMQ 增强类
 * @author Victor.Xiao
 * @since 2023-11-29 10:20
 **/
@Slf4j
public class RocketExtTemplate {

    private RocketMQTemplate template;

    private RocketMqConfigProperties rocketMqConfigProperties;

    public RocketExtTemplate(RocketMQTemplate template,RocketMqConfigProperties rocketMqConfigProperties) {
        this.template = template;
        this.rocketMqConfigProperties = rocketMqConfigProperties;
    }

    public RocketMQTemplate getTemplate() {
        return template;
    }

    /**
     * 根据环境重新隔离topic
     * @author Victor.Xiao
     * @since 2023-11-29 10:34
     * @param topic 原始topic
     * @return java.lang.String
    */
    public String rebuildTopic(String topic) {
        //手动去掉隔离配置,envIsolationFlag也可手动去掉隔离配置
        boolean envIsolationFlag = ContextUtil.get(ContextConstants.ROCKET_MQ_ENV_ISOLATION_FLAG, Boolean.class, true);
        //生产者是否环境隔离加一个开关
        if (envIsolationFlag && "1".equalsIgnoreCase(rocketMqConfigProperties.getEnvIsolationFlag()) && "1".equalsIgnoreCase(rocketMqConfigProperties.getProducerIsolationFlag()) ) {
            if (StringUtils.hasText(rocketMqConfigProperties.getRegion())) {
                topic = String.join("_", topic, rocketMqConfigProperties.getRegion());
            }
            if (StringUtils.hasText(rocketMqConfigProperties.getEnvironment())) {
                topic = String.join("_", topic, rocketMqConfigProperties.getEnvironment());
            }
        }
        return topic;
    }

    /**
     * 同步消息
     * @author Victor.Xiao
     * @since 2023-11-29 10:56
     * @param topic topic
     * @param message message
     * @return org.apache.rocketmq.client.producer.SendResult
    */
    public <T> SendResult syncSend(String topic, MsgDTO<T> message) {
        topic = rebuildTopic(topic);
        Message<MsgDTO<T>> sendMessage = buildMessage(message);
        SendResult sendResult = template.syncSend(topic, sendMessage);
        log.info("发送[{}] syncSend消息[{}] 发送结果[{}]", topic, message, sendResult);
        return sendResult;
    }

    /**
     * 异步消息
     * @author Victor.Xiao
     * @since 2023-11-29 10:56
     * @param topic topic
     * @param message message
     */
    public <T> void asyncSend(String topic, MsgDTO<T> message, SendCallback sendCallback) {
        topic = rebuildTopic(topic);
        Message<MsgDTO<T>> sendMessage = buildMessage(message);
        template.asyncSend(topic, sendMessage,sendCallback);
        log.info("发送[{}] asyncSend消息[{}]", topic, message);
    }

    /**
     * 发送单向消息
     * @author Victor.Xiao
     * @since 2023-11-29 10:56
     * @param topic topic
     * @param message message
     */
    public <T> void sendOneWay(String topic, MsgDTO<T> message) {
        topic = rebuildTopic(topic);
        Message<MsgDTO<T>> sendMessage = buildMessage(message);
        template.sendOneWay(topic, sendMessage);
        log.info("发送[{}] sendOneWay消息[{}]", topic, message);
    }


    /**
     * 延迟消息
     * messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
     * @author Victor.Xiao
     * @since 2023-11-29 11:13
     * @param topic topic
     * @param message message
     * @param delayLevel delayLevel
     * @return org.apache.rocketmq.client.producer.SendResult
    */
    public <T> SendResult sendDelay(String topic, MsgDTO<T> message, int delayLevel) {
        topic = rebuildTopic(topic);
        Message<MsgDTO<T>> sendMessage = buildMessage(message);
        //设置超时和延时推送
        SendResult sendResult = template.syncSend(topic, sendMessage, 3000, delayLevel);
        log.info("发送[{}] sendDelay消息[{}] ,delayLevel:[{}] 发送结果[{}]", topic, message, delayLevel, sendResult);
        return sendResult;
    }


    /**
     * 根据现有消息格式构建rocket消息对象
     * @author Victor.Xiao
     * @since 2023-11-29 11:02
     * @param message message
     * @return org.springframework.messaging.Message<com.ubxtech.core.dto.MsgDTO<T>>
    */
    private <T> Message<MsgDTO<T>> buildMessage(MsgDTO<T> message) {
        Message<MsgDTO<T>> sendMessage = null;
        if (StrUtil.isNotBlank(message.getId())) {
            sendMessage = MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.KEYS, message.getId()).build();
        } else {
            sendMessage = MessageBuilder.withPayload(message).build();
        }
        return sendMessage;
    }

}
